package org.study.es.api.client;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Result;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.mapping.*;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.json.JsonData;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import com.alibaba.fastjson2.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.study.es.api.model.User;

import java.io.IOException;
import java.util.*;

/**
 * 操作 {@link org.study.es.api.model.User} 的工具类
 *
 * @author doveylovey
 * @version v1.0.0
 * @email 1135782208@qq.com
 * @date 2023年06月03日
 */
@Slf4j
public class RestClient4User {
    /**
     * documentMapping写法，参数与实际的es脚本基本一致
     *
     * @param
     * @return java.util.Map<java.lang.String, co.elastic.clients.elasticsearch._types.mapping.Property>
     */
    public static Map<String, Property> createDocumentMapping() {
        Map<String, Property> documentMapping = Maps.newHashMap();
        documentMapping.put("name", Property.of(property -> property.text(TextProperty.of(textProperty -> textProperty.index(true).analyzer("ik_max_word")))));
        documentMapping.put("age", Property.of(property -> property.integer(IntegerNumberProperty.of(integerNumberProperty -> integerNumberProperty.index(true)))));
        return documentMapping;
    }

    /**
     * 添加文档
     * index()中的参数为文档所属的索引名
     * id()的中参数为文档的id
     * document()的参数为文档数据，新版本支持直接传入对象
     *
     * @param indexName
     * @param user
     * @return co.elastic.clients.elasticsearch._types.Result
     */
    public static Result addUser(String indexName, User user) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        // 向user对象中添加数据
        //User user = new User("java客户端", "男", 18);
        /*CreateRequest<Object> createRequest = CreateRequest.of(builder -> builder.index(indexName)
                .id(user.getUserId() + "")
                .document(user));
        CreateResponse response = client.create(createRequest);*/
        // 向索引中添加数据
        CreateResponse response = client.create(builder -> builder.index(indexName)
                .id(user.getUserId() + "")
                .document(user));
        Result result = response.result();
        log.info("向索引：{}中添加文档：{}，结果：{}", indexName, JSON.toJSONString(user), result);
        return result;
    }

    /**
     * 查询文档：如果查不到则会抛出 NullPointerException
     *
     * @param indexName
     * @param userId
     * @return org.study.es.api.model.User
     */
    public static User getUser(String indexName, String userId) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        GetResponse<User> response = client.get(builder -> builder.index(indexName).id(userId), User.class);
        User user = response.source();
        log.info("在索引：{}中查询userId={}的文档，结果：{}", indexName, userId, JSON.toJSONString(user));
        return user;
    }

    /**
     * 判断文档是否存在：如果查不到则会抛出 NullPointerException
     *
     * @param indexName
     * @param userId
     * @return boolean
     */
    public static boolean existsUser(String indexName, String userId) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        BooleanResponse response = client.exists(builder -> builder.index(indexName).id(userId));
        boolean value = response.value();
        log.info("在索引：{}中是否存在userId={}的文档：{}", indexName, userId, value);
        return value;
    }

    /**
     * 更新文档
     *
     * @param indexName
     * @param user
     * @return co.elastic.clients.elasticsearch._types.Result
     */
    public static Result updateUser(String indexName, User user) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        // 使用map集合封装需要修改的内容
        // Map<String, Object> map = new HashMap<>(2);
        // map.put("name", "java客户端aaa");
        UpdateResponse<User> updateResponse = client.update(builder -> builder.index(indexName)
                .id(user.getUserId() + "")
                .doc(user), User.class);
        Result result = updateResponse.result();
        log.info("修改索引：{}中的文档：{}，结果：{}", indexName, JSON.toJSONString(user), result);
        return result;
    }

    /**
     * 删除文档
     *
     * @param indexName
     * @param userId
     * @return co.elastic.clients.elasticsearch._types.Result
     */
    public static Result deleteUser(String indexName, String userId) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        DeleteResponse deleteResponse = client.delete(builder -> builder.index(indexName).id(userId));
        Result result = deleteResponse.result();
        log.info("删除索引：{}中的文档userId={}，结果：{}", indexName, userId, result);
        return result;
    }

    /**
     * 批量添加文档：核心是构建一个泛型为 BulkOperation 的 ArrayList 集合，实质上是将多个请求包装到一个集合中，进行统一请求，进行构建请求时调用 bulk() 方法，实现批量添加效果
     *
     * @param indexName
     * @param userList
     * @return java.util.List<co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem>
     */
    public static List<BulkResponseItem> addUserBatch(String indexName, List<User> userList) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        List<BulkOperation> list = new ArrayList<>();
        for (int i = 0; i < userList.size(); i++) {
            User user = userList.get(i);
            // BulkOperation operation = new BulkOperation.Builder().create(builder -> builder.index(indexName).document(user).id(user.getUserId() + "")).build();
            BulkOperation operation = BulkOperation.of(bulk -> bulk.index(index -> index.id(user.getUserId() + "").document(user)));
            list.add(operation);
        }
        BulkResponse bulkResponse = client.bulk(builder -> builder.index(indexName).operations(list));
        List<BulkResponseItem> itemList = bulkResponse.items();
        log.info("向索引：{}中的批量添加文档：{}，结果：{}", indexName, userList, itemList);
        return itemList;
    }

    /**
     * 批量修改文档：核心是构建一个泛型为 BulkOperation 的 ArrayList 集合，实质上是将多个请求包装到一个集合中，进行统一请求，进行构建请求时调用 bulk() 方法，实现批量添加效果
     *
     * @param indexName
     * @param userList
     * @return java.util.List<co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem>
     */
    public static List<BulkResponseItem> updateUserBatch(String indexName, List<User> userList) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        List<BulkOperation> list = new ArrayList<>();
        for (int i = 0; i < userList.size(); i++) {
            User user = userList.get(i);
            // BulkOperation operation = new BulkOperation.Builder().update(builder -> builder.index(indexName).id(user.getUserId() + "").action(param -> param.doc(user))).build();
            BulkOperation operation = BulkOperation.of(bulk -> bulk.update(index -> index.id(user.getUserId() + "").action(param -> param.doc(user))));
            list.add(operation);
        }
        BulkResponse bulkResponse = client.bulk(builder -> builder.index(indexName).operations(list));
        List<BulkResponseItem> itemList = bulkResponse.items();
        log.info("向索引：{}中的批量更新文档：{}，结果：{}", indexName, userList, itemList);
        return itemList;
    }

    /**
     * 批量删除文档：核心是构建一个泛型为 BulkOperation 的 ArrayList 集合，实质上是将多个请求包装到一个集合中，进行统一请求，进行构建请求时调用 bulk() 方法，实现批量添加效果
     *
     * @param indexName
     * @param userIdList
     * @return java.util.List<co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem>
     */
    public static List<BulkResponseItem> deleteUserBatch(String indexName, List<String> userIdList) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        // 构建一个批量数据集合
        List<BulkOperation> list = new ArrayList<>();
        for (int i = 0; i < userIdList.size(); i++) {
            String userId = userIdList.get(i);
            BulkOperation operation = new BulkOperation.Builder().delete(builder -> builder.index(indexName).id(userId)).build();
            // BulkOperation operation = BulkOperation.of(builder -> builder.delete(index -> index.id(userId)));
            list.add(operation);
        }
        BulkResponse bulkResponse = client.bulk(builder -> builder.index(indexName).operations(list));
        List<BulkResponseItem> itemList = bulkResponse.items();
        log.info("从索引：{}中的批量删除文档userIdList={}，结果：{}", indexName, userIdList, itemList);
        return itemList;
    }

    /**
     * 全量查询
     *
     * @param indexName
     * @return java.util.List<org.study.es.api.model.User>
     */
    public static List<User> searchAll(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .query(q -> q.matchAll(m -> m)), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        log.info("索引：{}中的文档总数：{}，结果：{}", indexName, total, userList);
        return userList;
    }

    /**
     * 分页查询：在全量查询的基础上增加了从第几条开始，查询几条
     *
     * @param indexName
     * @param pageNum
     * @param pageSize
     * @return java.util.Map<java.lang.String, java.lang.Object>
     */
    public static Map<String, Object> searchPage(String indexName, int pageNum, int pageSize) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .query(q -> q.matchAll(m -> m))
                .from((pageNum - 1) * pageSize)
                .size(pageSize), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        Map<String, Object> pageInfo = new LinkedHashMap<>(5) {{
            this.put("pageNum", pageNum);
            this.put("pageSize", pageSize);
            this.put("total", total);
            this.put("totalPage", (total % pageSize == 0) ? (total / pageSize) : (total / pageSize + 1));
            this.put("userList", userList);
        }};
        log.info("索引：{}分页查询结果：{}", indexName, pageInfo);
        return pageInfo;
    }

    /**
     * 排序查询
     *
     * @param indexName
     * @return java.util.List<org.study.es.api.model.User>
     */
    public static List<User> searchAgeAsc(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .query(q -> q.matchAll(m -> m))
                .sort(o -> o.field(f -> f.field("age").order(SortOrder.Desc))), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        log.info("索引：{}，总记录数：{}，排序查询结果：{}", indexName, total, userList);
        return userList;
    }

    /**
     * 条件查询：includes() 是显示的字段，excludes() 是排除的字段
     *
     * @param indexName
     * @return java.util.List<org.study.es.api.model.User>
     */
    public static List<User> searchColumn(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .query(q -> q.matchAll(m -> m))
                .sort(o -> o.field(f -> f.field("age").order(SortOrder.Asc)))
                .source(r -> r.filter(f -> f.includes("name", "age").excludes("sex"))), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        log.info("索引：{}，总记录数：{}，只查询指定字段结果：{}", indexName, total, userList);
        return userList;
    }

    /**
     * 组合查询：must 是必须满足所有条件，should 只要满足一个就行
     *
     * @param indexName
     * @return java.util.List<org.study.es.api.model.User>
     */
    public static List<User> searchComposite4Must(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .query(q -> q.bool(b -> b.must(m -> m.match(u -> u.field("age").query(18)))
                        .must(m -> m.match(u -> u.field("sex").query("男")))
                        .mustNot(m -> m.match(u -> u.field("sex").query("女"))))), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        log.info("索引：{}，总记录数：{}，多条件查询结果：{}", indexName, total, userList);
        return userList;
    }

    /**
     * 组合查询：must 是必须满足所有条件，should 只要满足一个就行
     *
     * @param indexName
     * @return java.util.List<org.study.es.api.model.User>
     */
    public static List<User> searchComposite4Should(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .query(q -> q.bool(b -> b.should(h -> h.match(u -> u.field("age").query(18)))
                        .should(h -> h.match(u -> u.field("sex").query("女"))))), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        log.info("索引：{}，总记录数：{}，多条件查询结果：{}", indexName, total, userList);
        return userList;
    }

    /**
     * 范围查询：gte()表示取大于等于，gt()表示大于，lte()表示小于等于
     *
     * @param indexName
     * @return java.util.List<org.study.es.api.model.User>
     */
    public static List<User> searchAgeRange(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .query(q -> q.range(r -> r.field("age").gte(JsonData.of(19)).lte(JsonData.of(21)))), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        log.info("索引：{}，总记录数：{}，范围条件查询结果：{}", indexName, total, userList);
        return userList;
    }

    /**
     * 模糊查询：fuzziness 表示差几个可以查询出来
     *
     * @param indexName
     * @return java.util.List<org.study.es.api.model.User>
     */
    public static List<User> searchFuzzy(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName).query(q ->
                q.fuzzy(f -> f.field("name").value("tsst-1").fuzziness("2"))), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        log.info("索引：{}，总记录数：{}，fuzzy查询结果：{}", indexName, total, userList);
        return userList;
    }

    /**
     * 高亮查询
     *
     * @param indexName
     * @return java.util.List<org.study.es.api.model.User>
     */
    public static List<User> searchHighlight(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .query(q -> q.term(t -> t.field("name").value("test")))
                .highlight(h -> h.fields("name", f -> f.preTags("<font color='red'>").postTags("</font>"))), User.class);
        HitsMetadata<User> hits = searchResponse.hits();
        List<User> userList = Lists.newArrayList();
        for (Hit<User> hit : hits.hits()) {
            userList.add(hit.source());
        }
        long total = hits.total().value();
        log.info("索引：{}，总记录数：{}，高亮查询结果：{}", indexName, total, userList);
        return userList;
    }

    /**
     * 聚合查询：取最大年龄
     *
     * @param indexName
     * @return java.util.Map<java.lang.String, co.elastic.clients.elasticsearch._types.aggregations.Aggregate>
     */
    public static Map<String, Aggregate> searchAggregations(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .aggregations("maxAge", a -> a.max(m -> m.field("age"))), User.class);
        Map<String, Aggregate> aggregateMap = searchResponse.aggregations();
        // aggregateMap.entrySet().forEach(f -> System.out.println(f.getKey() + ":" + f.getValue().max().value()));
        long total = searchResponse.hits().total().value();
        log.info("索引：{}，总记录数：{}，聚合查询(取最大年龄)结果：{}", indexName, total, aggregateMap);
        return aggregateMap;
    }

    /**
     * 分组查询：实质上是聚合查询的一种
     *
     * @param indexName
     * @return java.util.List<co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket>
     */
    public static List<LongTermsBucket> searchGroup(String indexName) throws IOException {
        ElasticsearchClient client = RestClientUtils.getElasticsearchClient();
        SearchResponse<User> searchResponse = client.search(builder -> builder.index(indexName)
                .aggregations("ageGroup", a -> a.terms(t -> t.field("age"))), User.class);
        List<LongTermsBucket> longTermsBucketList = searchResponse.aggregations().get("ageGroup").lterms().buckets().array();
        // longTermsBucketList.forEach(f -> System.out.println(f.key() + ":" + f.docCount()));
        long total = searchResponse.hits().total().value();
        log.info("索引：{}，总记录数：{}，分组查询结果：{}", indexName, total, longTermsBucketList);
        return longTermsBucketList;
    }
}
